package com.idanchuang.support.etcache.pubsub;

import com.idanchuang.support.etcache.exception.InvalidRemoteKeyException;
import com.idanchuang.support.etcache.remote.RemoteKey;
import com.idanchuang.support.etcache.remote.codec.RemoteKeyCodec;
import io.lettuce.core.AbstractRedisClient;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
import io.lettuce.core.cluster.pubsub.RedisClusterPubSubAdapter;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
import io.lettuce.core.cluster.pubsub.api.sync.RedisClusterPubSubCommands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceClusterConnection;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.serializer.RedisSerializer;

import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 基于Redis Lettuce实现key过期事件监听, 并将事件push至SubEventQueue
 *
 * @author yjy
 * Created at 2021/12/22 12:44 下午
 */
public class RedisRemoteExpireSubscriber extends RedisClusterPubSubAdapter<String, String> implements MessageListener {

    private static final Logger logger = LoggerFactory.getLogger(RedisRemoteExpireSubscriber.class);
    private final ExecutorService processor = new ThreadPoolExecutor(3, 3,
            1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(),
            (r -> new Thread(r, "et-cache-worker")));

    private static final String EXPIRE_EVENT_PATTERN = "__keyevent@*__:expired";

    private final RemoteKeyCodec remoteKeyCodec;
    private final SubEventQueue subEventQueue;

    public RedisRemoteExpireSubscriber(RedisConnectionFactory redisConnectionFactory, RemoteKeyCodec remoteKeyCodec, SubEventQueue subEventQueue) throws IllegalAccessException, NoSuchFieldException {
        this.subEventQueue = subEventQueue;
        this.remoteKeyCodec = remoteKeyCodec;
        RedisConnection connection = redisConnectionFactory.getConnection();
        if (connection instanceof LettuceClusterConnection) {
            LettuceConnectionFactory factory = (LettuceConnectionFactory) redisConnectionFactory;
            Field field = LettuceConnectionFactory.class.getDeclaredField("client");
            field.setAccessible(true);
            AbstractRedisClient client = (AbstractRedisClient) field.get(factory);
            if (client instanceof RedisClusterClient) {
                StatefulRedisClusterPubSubConnection<String, String> pubSubConnection = ((RedisClusterClient) client).connectPubSub();
                pubSubConnection.addListener(this);
                pubSubConnection.setNodeMessagePropagation(true);
                RedisClusterPubSubCommands<String, String> sync = pubSubConnection.sync();
                sync.masters().commands().psubscribe(EXPIRE_EVENT_PATTERN);
            } else {
                throw new UnsupportedOperationException("Can not resolve redis pubsub with: " + client.getClass().getName());
            }
        } else {
            byte[] patternBytes = RedisSerializer.string().serialize(EXPIRE_EVENT_PATTERN);
            connection.pSubscribe(this, patternBytes);
        }
    }

    @Override
    public void message(RedisClusterNode node, String pattern, String channel,
                        String message) {
       handle(message.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
       handle(message.getBody());
    }

    /**
     * 解析事件消息, 并将事件push至subEventQueue
     * @param remoteKeyBytes 事件消息
     */
    private void handle(byte[] remoteKeyBytes) {
        processor.execute(() -> {
            try {
                RemoteKey remoteKey = remoteKeyCodec.decode(remoteKeyBytes);
                RemoteEvent remoteEvent = new RemoteEvent(remoteKey.getCn(), remoteKey);
                subEventQueue.push(remoteEvent);
            } catch (InvalidRemoteKeyException ignored) {
            } catch (Exception e) {
                logger.error("handle failed", e);
            }
        });
    }

}
